package com.gitee.xhs.inbuiltWatermarkGenerator;

import com.gitee.xhs.EnvironmentConfig;
import com.gitee.xhs.FlinkEnvironmentFactory;
import com.gitee.xhs.WatermarkRecordMakerProcessFunction;
import com.gitee.xhs.WatermarkSourceFactory;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.time.Duration;

import static com.gitee.xhs.WatermarkConstant.DEFAULT_WATERMARK_STRATEGY;
import static com.gitee.xhs.WatermarkConstant.*;

/**
 * @author BigHead
 * @date 2022/7/15 14:31
 * @description BoundedOutOfOrdernessMain
 * <p>
 * 通过设置次参数 getConfig().setAutoWatermarkInterval(Long.MAX_VALUE)
 * 调整自动时间产生结果，注释掉可以看到相应效果，具体源码在
 * @see org.apache.flink.api.common.eventtime.BoundedOutOfOrdernessWatermarks
 */
public class BoundedOutOfOrdernessMain {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment streamExecutionEnvironment = FlinkEnvironmentFactory.webUi(new EnvironmentConfig(null));

        streamExecutionEnvironment.getConfig().setAutoWatermarkInterval(Long.MAX_VALUE);

        streamExecutionEnvironment
                .addSource(
                        WatermarkSourceFactory.defaultTransactionBoundOneSecondSource()
                )
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy
                                .forBoundedOutOfOrderness(Duration.ofSeconds(10))
                )
                .process(
                        new WatermarkRecordMakerProcessFunction(DEFAULT_WATERMARK_STRATEGY, INBUILT_BOUNDED_OUT_OF_ORDERNESS, RECORD_TIMESTAMP_ASSIGNER)
                )
                .print();


        streamExecutionEnvironment.execute("CustomWaterMarkMain");
    }


}
